package com.roocon.thread.chapter02;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.stream.IntStream;

/**
 * 手写线程池
 */
public class SimpleThreadPool extends Thread {

  private final static int DEFAULT_TASK_QUEUE_SIZE = 2000;

  private int size;

  private static int seq = 0;

  private final static String THREAD_PREFIX = "SIMPLE_THREAD_POOL-";

  private final static ThreadGroup GROUP = new ThreadGroup("SIMPLE_GROUP");

  private final static LinkedList<Runnable> TASK_QUEUE = new LinkedList<>();

  private final static LinkedList<WorkerTask> THREAD_QUEUE = new LinkedList<>();

  private volatile boolean destroy = false;

  private int min;

  private int max;

  private int active;

  public final static DiscardPolicy DEFAULT_DISCARD_POLICY = () -> {
    throw new DiscardException("Discard for the pool is overflow !");
  };

  public static class DiscardException extends RuntimeException {
    public DiscardException(String message) {
      super(message);
    }
  }

  private final DiscardPolicy discardPolicy;

  private int queueSize;

  public SimpleThreadPool(int min, int active, int max, int queueSize,
      DiscardPolicy discardPolicy) {
    this.min = min;
    this.max = max;
    this.active = active;
    this.queueSize = queueSize;
    this.discardPolicy = discardPolicy;
    init();
  }

  public SimpleThreadPool() {
    this(4, 8, 12, DEFAULT_TASK_QUEUE_SIZE, DEFAULT_DISCARD_POLICY);
  }

  private void init() {
    for (int i = 0; i < min; i++) {
      createWorkerTask();
    }
    this.size = min;
    start();
  }

  @Override public void run() {
    while (!isDestroy()) {
      System.out.println(toString());
      try {
        Thread.sleep(5_000L);
        if (TASK_QUEUE.size() > active && size < active) {
          for (int i = size; i < active; i++) {
            createWorkerTask();
          }
          size = active;
          System.out.println("The pool incremented to active");
        } else if (TASK_QUEUE.size() > max && size < max) {
          for (int i = size; i < max; i++) {
            createWorkerTask();
          }
          size = max;
          System.out.println("The pool increment to max num. ");
        }

        if (TASK_QUEUE.isEmpty() && size > active) {
          System.out.println("============reduce===========");
          synchronized (TASK_QUEUE) {
            int releaseSize = size - active;
            for (Iterator<WorkerTask> it = THREAD_QUEUE.iterator(); it.hasNext();) {
              if (releaseSize <= 0) break;

              WorkerTask task = it.next();
              task.close();
              task.interrupt();
              it.remove();
              releaseSize --;
            }
            size = active;
          }
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }

  private void createWorkerTask() {
    WorkerTask task = new WorkerTask(GROUP, THREAD_PREFIX + (seq++));
    task.start();
    THREAD_QUEUE.add(task);
  }

  public void shutDown() throws InterruptedException {
    while (!TASK_QUEUE.isEmpty()) {
      Thread.sleep(50);
    }
    int initVal = THREAD_QUEUE.size();
    while (initVal > 0) {
      for (WorkerTask task : THREAD_QUEUE) {
        if (task.taskState == TaskState.BLOCKED) {
          task.interrupt(); // 处理等待状态
          task.close(); // 处理运行状态
          initVal--;
        } else {
          Thread.sleep(10);
        }
      }
    }
    this.destroy = true;
    System.out.println("This pool has been shutdown successfully");
  }

  public int getSize() {
    return size;
  }

  public int getQueueSize() {
    return queueSize;
  }

  public int getMin() {
    return min;
  }

  public int getMax() {
    return max;
  }

  public int getActive() {
    return active;
  }

  public boolean isDestroy() {
    return destroy;
  }

  public void submit(Runnable runnable) {
    if (destroy) {
      throw new IllegalArgumentException(
          "The thread pools already destroy and not allow submit task.");
    }

    synchronized (TASK_QUEUE) {
      if (TASK_QUEUE.size() > queueSize) discardPolicy.discard();
      TASK_QUEUE.addLast(runnable);
      TASK_QUEUE.notifyAll();
    }
  }

  public interface DiscardPolicy {
    void discard() throws DiscardException;
  }

  private enum TaskState {
    FREE, RUNNING, BLOCKED, DEAD
  }

  @Override public String toString() {
    return "SimpleThreadPool{" +
        "size=" + size +
        ", destroy=" + destroy +
        ", min=" + min +
        ", max=" + max +
        ", active=" + active +
        //", discardPolicy=" + discardPolicy +
        ", queueSize=" + queueSize +
        '}';
  }

  private static class WorkerTask extends Thread {

    private volatile TaskState taskState = TaskState.FREE;

    public WorkerTask(ThreadGroup group, String name) {
      super(group, name);
    }

    public TaskState getTaskState() {
      return taskState;
    }

    public void close() {
      this.taskState = TaskState.DEAD;
    }

    public void run() {
      OUTER:
      while (this.taskState != TaskState.DEAD) {
        Runnable runnable;
        synchronized (TASK_QUEUE) {
          while (TASK_QUEUE.isEmpty()) {
            try {
              taskState = TaskState.BLOCKED;
              TASK_QUEUE.wait();
            } catch (InterruptedException e) {
              System.out.println("Closed.");
              break OUTER;
            }
          }
          runnable = TASK_QUEUE.removeFirst();
          // IMP: 2019/4/7  注意 synchronized 里面和外面的区别
          //if (runnable != null) {
          //  taskState = TaskState.RUNNING;
          //  runnable.run();
          //  taskState = TaskState.FREE;
          //}
        }
        if (runnable != null) {
          taskState = TaskState.RUNNING;
          runnable.run();
          taskState = TaskState.FREE;
        }
      }
    }
  }

  public static void main(String[] args) throws InterruptedException {
    SimpleThreadPool pool = new SimpleThreadPool();
    IntStream.rangeClosed(0, 40)
        .forEach(i -> {
          pool.submit(() -> {
            System.out.println(
                "The runnable [" + i + "]be services by " + Thread.currentThread() + "start");
            try {
              Thread.sleep(10_000);
            } catch (InterruptedException e) {
              e.printStackTrace();
            }
            System.out.println(
                "The runnable [" + i + "]be services by " + Thread.currentThread() + "finish !");
          });
        });

    //Thread.sleep(10_000);
    //pool.shutDown();
    //pool.submit(()->{
    //  System.out.println("==========");
    //});
  }
}
